"""
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    created by lane.chang on '07/09/2022'
    comment: 封装的最底层函数
"""
import os
import re
import random
import string
import datetime
import asyncio
import threading
import math
from decimal import Decimal
from importlib import import_module
from typing import Any, Optional, Union
from inspect import getmembers
from urllib.parse import urlparse
from enum import Enum
from pydantic import BaseModel, Field

import pypinyin
from starlette_context import context
from fastapi.responses import JSONResponse

from project.core.exception import DEFAULT_ERROR_CODES


# 数字汉字
NUM_CHR = {
    0: '零',
    1: '一',
    2: '二',
    3: '三',
    4: '四',
    5: '五',
    6: '六',
    7: '七',
    8: '八',
    9: '九',
    10: '十'
}

# 全国省/市编号列表
province_codes = {
    '11', '12', '13', '14', '15', '21', '22', '23', '31', '32',
    '33', '34', '35', '36', '37', '41', '42', '43', '44', '45',
    '46', '50', '51', '52', '53', '54', '61', '62', '63', '64',
    '65', '71', '81', '82', '91'
}


def get_request_id() -> str:
    return context.data["X-Request-ID"]


def get_process_time() -> float:
    # return eval(context.data["X-Process-Time"]) * 10**6
    return 1


class ResponseModel(BaseModel):
    """API返回对象"""

    code: int = Field(default=0, description="状态码")
    message: str = Field(default="success", description="消息描述")
    # request_id: str = Field(default_factory=get_request_id, description="")
    data: Optional[Union[dict, list, Any]] = Field(
        default=None, description="数据内容"
    )
    exec_time: float = Field(default=0.0, description="逻辑执行时长(ms)")


def handler_text(text: str, length: int, remain_eof: bool = True) -> str:
    """对超长文本进行缩略处理
    :param text:
    :param length:
    :param remain_eof: 是否保留原结尾符
    :return:
    """
    if not text:
        return text
    if len(text) <= length:
        return text

    if remain_eof:
        text = text.replace(text[length:-1], "...")
    else:
        text = text.replace(text[length:], "...")
    return text


def serializer(
    data: Optional[Union[tuple, list, set,
                         dict]]) -> Optional[Union[dict, list]]:
    """递归序列化: 有几种基本类型, 比如时间,日期, decimal 需要序列化
    :param data:
    :return:
    """
    if data is None:
        return None

    # 如果是list或者tuple set
    if isinstance(data, (tuple, list, set)):
        dicts = []
        for each in data:
            dicts.append(serializer(each))
        return dicts
    # 不是dict 就直接返回
    if not isinstance(data, dict):
        return data
    # 将字典中的各种不能序列号的类型转化成字符串
    result_data = {}
    for key, value in data.items():
        if isinstance(value, (tuple, list, dict)):
            # 如果值是list, tuple, dict, 那么需要遍历下去
            result_data[key] = serializer(value)
        elif isinstance(value, datetime.datetime):
            result_data[key] = value.strftime("%Y-%m-%d %H:%M:%S")
            result_data[key + '_' + 'ts'] = int(value.timestamp())
        elif isinstance(value, datetime.date):
            result_data[key] = value.strftime("%Y-%m-%d")
        elif isinstance(value, datetime.time):
            result_data[key] = value.strftime("%H:%M:%S")
        elif isinstance(value, Decimal):
            result_data[key] = str(value)
        elif isinstance(value, Enum):
            result_data[key] = value.value
        else:
            result_data[key] = value

    return result_data


def json_response(
    code: int = 0,
    data: Optional[Union[tuple, list, set, dict, str, bool]] = None,
    message: Optional[str] = None,
) -> JSONResponse:
    """构造api的json返回值
    :param code:
    :param data:
    :param message:
    :return:
    """
    response_model = ResponseModel()

    response_model.code = code

    msg_head = DEFAULT_ERROR_CODES.get(code)
    if message:
        response_model.message = message
    elif msg_head:
        response_model.message = msg_head

    if data is not None:
        response_model.data = serializer(data)

    return JSONResponse(content=response_model.dict())


def get_modules(package_name: str) -> list[str]:
    """获取包名下所有非__init__的模块名
    :param: package_name: 指定包名(含导入路径名)
    """
    package = import_module(package_name)
    path_name = package.__path__[0]
    files = os.listdir(path_name)
    modules = []
    for file in files:
        if file.startswith("__"):
            continue
        if os.path.isfile(f"{path_name}/{file}"):
            if not file.endswith(".py"):
                continue
            name, _ = os.path.splitext(file)
            if not package_name:
                modules.append(name)
            else:
                modules.append(f"{package_name}.{name}")
        else:
            modules.extend(get_modules(package_name=f"{package_name}.{file}"))

    return modules


def get_models(package_name: str, predicate=None) -> list[Any]:
    """获取包名下所有指定类
    :param: package_name: 指定包名(含导入路径名)
    """
    modules = get_modules(package_name)
    ret_models = []
    for module in modules:
        _module = import_module(module)
        ret_models.extend(getmembers(_module, predicate))

    return [v for _, v in ret_models]


def handler_text(text: str, length: int, remain_eof: bool = True) -> str:
    """对超长文本进行缩略处理
    :param text:
    :param length:
    :param remain_eof: 是否保留原结尾符
    :return:
    """
    if not text:
        return text
    if len(text) <= length:
        return text

    if remain_eof:
        text = text.replace(text[length:-1], "...")
    else:
        text = text.replace(text[length:], "...")
    return text


def generate_path(file_path: str = ""):
    """检查并生成生成文件路径
    :param file_path:
    :return:
    """
    if not os.path.exists(file_path):
        os.makedirs(file_path)

    return file_path


def generate_random_letter():
    """生成随机字母
    :return:
    """
    return random.choice(string.ascii_lowercase)


def get_login_key():
    """ 根据人员类型, 返回不同的login_key
    :return:
    """
    return 'login_token_user'


def set_login_info(rds, user_id, token):
    """
    :desc: 设置登录用户的信息(含刷新)
    :return:
    """
    rds.hset(name=get_login_key(), key=user_id, value=token)


def delete_login_info(rds, user_id):
    """
    :desc: 删除用户登陆信息
    :return:
    """
    rds.hdel(get_login_key(), user_id)


def convert_py(text: Union[Any], separator: str = '', ret_form: str = 'str', style=pypinyin.NORMAL) -> Union[str, list]:
    """ 转拼音
    :param: text: 文本
    :param: separator: 分隔符
    :param: ret_form: 返回形式
    :param: style: 拼音类型 normal(无声调)/tone(有声调)
    :return:
    """
    try:
        if type(text) != str:
            return text
        text_py = pypinyin.pinyin(text, style=style)
        if ret_form == 'str':
            return separator.join([v[0] for v in text_py])

        if ret_form == 'list':
            return [v[0] for v in text_py]
    except Exception as ex:
        print(ex)
        return text


def pop_elements(_list, count=1, is_random=True):
    """
    抛出指定数量的元素

    Args:
        _list(list): 列表
        count(int): 抛出元素的位数量
        is_random(bool): 是否随机

    Raises:
        IndexError

    Returns:
        列表抛出的结果
    """
    # 判断是否为空
    if not _list:
        return
    if len(_list) < count:
        count = len(_list)
    # 是否随机
    if is_random:
        return [_list.pop(random.choice(range(0, len(_list)))) for _ in range(count)]

    return [_list.pop(-1) for _ in range(count)]


def start_new_thread() -> asyncio.AbstractEventLoop:
    """ 创建并启动新线程 - 实现异步
    :return:
    """
    new_loop: asyncio.AbstractEventLoop = asyncio.new_event_loop()

    def loop_in_thread(_loop):  # 一个将被丢进线程的函数
        asyncio.set_event_loop(_loop)  # 调用loop需要使用set_event_loop方法指定loop
        _loop.run_forever()  # run_forever() 会永远阻塞当前线程，直到有人停止了该loop为止。

    # 开启客户端接收
    t = threading.Thread(target=loop_in_thread, args=(new_loop,))
    t.start()

    return new_loop


def format_seconds(value: int) -> str:
    """ 格式化秒
    :return:
    """
    ret_str = ''
    # 时
    hour = value // 3600
    if hour > 0:
        ret_str += f'{hour}时'

    # 分
    minutes = (value - hour * 3600) // 60
    if minutes > 0:
        ret_str += f'{minutes}分'

    # 秒
    seconds = value - hour * 3600 - minutes * 60
    if seconds > 0:
        ret_str += f'{seconds}秒'

    if not ret_str:
        ret_str = '0秒'

    return ret_str


def is_url_urllib(str_value):
    """
    判断字符串是否为URL的urllib库方法
    """
    try:
        result = urlparse(str_value)
        return all([result.scheme, result.netloc])
    except ValueError:
        return False


def num_convert_chr(value: str) -> str:
    """ 数值转汉字
    :return:
    """
    ret = ''
    for v in value:
        if v.isdigit():
            ret += NUM_CHR.get(int(v))
        else:
            ret += v

    return ret


def is_subsequence(content: Union[list, str], text: Union[list, str]):
    """ 是否按顺序包含
    :param content: 小文本
    :param text: 大文本
    :return:
    """
    text = iter(text)
    return all(v in text for v in content)


def match_index(text: Union[str, list, tuple], content: Union[str, int]) -> int:
    """ 返回匹配内容在文本中的位置
    :param text: 大文本
    :param content:
    :return: 
    """
    try:
        index = text.index(content)
    except ValueError as ex:
        pass
        index = -1

    return index


def cal_age_with_birth_ym(birth_ym: str):
    """
    :param birth_ym:
    :return:
    """
    start_date = None
    if len(birth_ym) == 7:
        start_date = datetime.datetime.strptime(birth_ym, '%Y-%m')
    elif len(birth_ym) == 6:
        start_date = datetime.datetime.strptime(birth_ym, '%Y%m')

    age = 0
    if start_date:
        end_date = datetime.datetime.now()
        months = (end_date.year - start_date.year) * 12 + end_date.month - start_date.month
        age = math.floor(months / 12) if months > 0 else 0

    return age


def get_intersection_text(text_compared: str, text: str, exact=True, style=pypinyin.NORMAL, mode='complete') -> dict:
    """ 返回文本交集匹配的部分(从字符串的开始部分匹配)
    :param text_compared: 要对照的文本
    :param text:
    :param exact: 是否精确匹配(主要包括拼音匹配逻辑)
    :param style:
    :param mode:
    :return:
    """
    match_text = ''
    text_updated = text_compared
    match_index = -1
    ret_data = {
        'match_text': match_text,
        'text_updated': text_updated,
        'match_index': match_index
    }
    if exact:
        count = 0
        while True:
            if count == 0:
                text_tmp = text
            else:
                text_tmp = text[0: count]

            if text_tmp in text_compared:
                match_text = text_tmp
                match_index = text_compared.index(text_tmp)
                text_updated = text_compared.replace(match_text, '')
                break

            if int(math.fabs(count)) >= len(text) - 1:
                break

            count -= 1
    else:
        _texts_compared = convert_py(num_convert_chr(text_compared), ret_form='list', style=style)
        texts = convert_py(num_convert_chr(text), ret_form='list', style=style)

        intersection_list, start_index = get_intersection_list(texts_compared=_texts_compared, texts=texts)

        for index in range(len(intersection_list)):
            match_text += text[index]
            match_index = start_index

        text_updated = text_compared.replace(match_text, '')

    if mode == 'complete' and match_text != text:
        return ret_data

    ret_data['match_text'] = match_text
    ret_data['text_updated'] = text_updated
    ret_data['match_index'] = match_index

    return ret_data


def get_intersection_list(texts_compared: list[Any], texts: list[Any]) -> (list[Any], int):
    """ 返回列表交集匹配部分
    :param texts_compared: 要对照的列表
    :param texts:
    :return:
    """
    # if len(texts) > len(texts_compared):
    #     return []

    match_indexes = []
    texts_compared_tmp = texts_compared
    index_tmp = 0
    while True:
        text = texts[0]

        index = match_index(texts_compared_tmp, text)
        if index == -1:
            break

        match_indexes.append(index_tmp + index)
        index_tmp += index + 1
        texts_compared_tmp = texts_compared_tmp[index + 1:]

    match_texts = []
    start_index = -1
    num = 0
    for index in match_indexes:
        text_compared = texts_compared[index]
        text = texts[0]
        count = 1
        while True:
            if index + count > len(texts_compared) - 1:
                count -= 1
                break
            text_compared += texts_compared[index + count]

            if count > len(texts) - 1:
                count -= 1
                break
            text += texts[count]

            if text_compared != text:
                count -= 1
                break
            count += 1

        if len(match_texts) <= count + 1:
            match_texts = texts[0: count + 1]
            start_index = index
            num = count + 1

    if start_index > -1:
        for i in range(start_index + num - 1, start_index - 1, -1):
            texts_compared.pop(i)

    return match_texts, start_index


def get_match_indexes(content: Union[list, str], text: Union[list, str]) -> Optional[list[int]]:
    """ 返回文本公共部分index列表
    :param content: 小文本
    :param text: 大文本
    :return:
    """
    if not is_subsequence(content, text):
        return

    indexes = []
    for index, v in enumerate(text):
        if v in content:
            indexes.append(index)

        if match_index(content, v) >= len(content) - 1:
            break

    return indexes


# 判断字符是否为汉字
def is_chinese(word):
    """ 判断是否为汉字
    :param word:
    :return:
    """
    pattern = re.compile(r'[^\u4e00-\u9fa5]')
    if pattern.search(word):
        return False
    else:
        return True


def validate_id_number(id_number: str) -> bool:
    """ 验证身份证号码是否合法
    :param id_number:
    :return:
    """
    if not id_number:
        return False
    # 验证身份证号码格式是否正确
    pattern = r'^\d{17}[\dXx]$'
    if not re.match(pattern, id_number):
        return False

    # 验证身份证号码中的地区码是否正确
    province_code = id_number[:2]
    if province_code not in province_codes:
        return False

    # 验证身份证号码中的出生日期是否正确
    birthday = id_number[6:14]
    try:
        datetime.datetime.strptime(birthday, '%Y%m%d')
    except ValueError:
        return False

    # 验证身份证号码中的校验码是否正确
    factors = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
    check_sum = sum(int(id_number[i]) * factors[i] for i in range(17))
    check_code = '10X98765432'[check_sum % 11]
    if check_code != id_number[-1].upper():
        return False

    # 身份证号码验证通过
    return True


def if_none(value: Any, ret_value: Union[str, int] = '') -> Union[str, int]:
    """ 如果为None, 则返回指定的字符串
    :param value:
    :param ret_value: 指定的
    :return:
    """
    return ret_value if value is None else value


def validate_phone(phone: str) -> bool:
    """ 验证身份证号码是否合法
    :param phone:
    :return:
    """
    pattern = re.compile(r'^(13[0-9]|14[0|5|6|7|9]|15[0|1|2|3|5|6|7|8|9]|'
                         r'16[2|5|6|7]|17[0|1|2|3|5|6|7|8]|18[0-9]|'
                         r'19[1|3|5|6|7|8|9])\d{8}$')

    return bool(re.match(pattern, phone))


def camel_to_snake(value):
    """ 字符串驼峰规则转长蛇规则
    :param value:
    :return:
    """
    result = ''
    for i in range(len(value)):
        if value[i].isupper():
            result += '_' + value[i].lower()
        else:
            result += value[i]

    return result.lstrip('_')


def split_text(text, length=30):
    """ 根据长度截取字符串
    :param text:
    :param length:
    :return:
    """
    texts = []
    for i in range(0, len(text), length):
        texts.append(text[i:i + length])

    return texts


def extract_phone_numbers(text):
    """ 从文本中筛选出手机号信息
    :param text:
    :return:
    """
    pattern = r'1[3456789]\d{9}'

    phone_numbers = re.findall(pattern, text)
    return phone_numbers


def to_dict(exclude_unset=False, **kwargs):
    """
    :param kwargs:
    :param exclude_unset:
    :return:
    """
    if not exclude_unset:
        return kwargs
    else:
        ret_kwargs = dict()
        for k, v in kwargs.items():
            if v is None:
                continue
            ret_kwargs[k] = v

        return ret_kwargs


class Core(BaseModel):
    """
    """
    func: Any = Field(..., title='异步函数')
    name: str = Field(..., title='函数名')


async def async_run(*args):
    """ 异步执行任务
    :param args:
        core:
        name:
    :return:
    """
    ret_data = dict()
    tasks = list()
    for core in args:
        tasks.append(asyncio.create_task(core.func, name=core.name))

    await asyncio.gather(*tasks)

    for task in tasks:
        ret_data[task.get_name()] = task.result()

    return ret_data


def is_iterable(obj):
    try:
        iter(obj)
        return True
    except TypeError:
        return False


def is_subclass(cls_sub, cls):
    """
    :param cls_sub:
    :param cls:
    :return:
    """
    try:
        return issubclass(cls_sub, cls)
    except TypeError:
        return False

